package com.heima.article.stream;

import com.alibaba.fastjson.JSON;
import com.heima.common.constants.HotArticleConstants;
import com.heima.model.mess.ArticleVisitStreamMess;
import com.heima.model.mess.UpdateArticleMess;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;
import java.util.Arrays;

@Configuration
@Slf4j
public class KafkaStreamHelloListener {

    @Bean
    public KStream<String,String> kStream(StreamsBuilder streamsBuilder){
        KStream<String, String> source = streamsBuilder.stream(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC);
        source
                //流式处理
                //按照key分组 文章id
                .groupByKey()
                .windowedBy(TimeWindows.of(Duration.ofSeconds(5)))
                .aggregate(() -> JSON.toJSONString(new ArticleVisitStreamMess()),
                        (key, value, agg) -> {
                            //每一个消息过来的时候，进行累加动作

                            //agg 反序列化成 ArticleVisitStreamMess
                            //上一次累加完结果
                            ArticleVisitStreamMess mess =
                                    JSON.parseObject(agg, ArticleVisitStreamMess.class);
                            //文章ID
                            Long articleId = Long.valueOf(key);
                            //消息（点赞1次...）
                            UpdateArticleMess updateArticleMess = JSON.parseObject(value, UpdateArticleMess.class);

                            switch (updateArticleMess.getType()){
                                case COMMENT:
                                    mess.setComment(mess.getComment() + updateArticleMess.getAdd());
                                    break;
                                case VIEWS:
                                    mess.setView(mess.getView() + updateArticleMess.getAdd());
                                    break;
                                case LIKES:
                                    mess.setLike(mess.getLike() + updateArticleMess.getAdd());
                                    break;
                                case COLLECTION:
                                    mess.setCollect(mess.getCollect() + updateArticleMess.getAdd());
                                    break;
                                default:break;
                            }

                            if(mess.getArticleId() == null){
                                mess.setArticleId(articleId);
                            }

                            return JSON.toJSONString(mess);
                        }
                )
                .toStream()
                .map((KeyValueMapper<Windowed<String>, String, KeyValue<?, ?>>)
                        (stringWindowed, s) ->
                                new KeyValue<>(stringWindowed.key(),s))
                .to(HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC);
        return source;
    }
}